package com.example.bigdata.spark.Transformations算子

import org.apache.spark.Partitioner.defaultPartitioner
import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

import scala.reflect.ClassTag
import scala.util.control.Breaks._
object TransformationsDemo {
    def main(args: Array[String]): Unit = {
        val sc = sparkContextInit()
        mapDemo(sc)     // 遍历RDD每个数据，产生新的RDD
        flatMapDemo(sc)     // 对每个元素通过F函数生成新的RDD
        filterDemo(sc)      // 对每个元素通过f函数过滤出true 的元素，生成新RDD
        distinctDemo(sc)    // 对RDD 进行去重
        mapPartitionsDemo(sc)   // 按照分区，对分区元素进行Map操作
        mapPartitionsWithIndexDemo(sc)  //显示数据和分区数
        unionDemo(sc)   // RDD进行并集运算
        intersectionDemo(sc)    //RDD进行交集运算
        subtractDemo(sc)    // RDD进行差集运算 ， 不去重
        coalesceDemo(sc)    // 修改RDD 分区数
        repartitionDemo(sc) // 修改RDD 分区数
        randomSplitDemo(sc) // 对RDD 按照权重进行拆分 拆分的新RDD 个数与权重有关
        glomDemo(sc)    // 将分区数据变成数组，元素数据类型不变；查看同分区内数据
        zipDemo(sc)     // 对RDD1[(k1,k2,k3)],RDD2[(v1,v2,v3)]进行合并 ,获得新的RDD[(k1,v1),(k2,v2),(k3,v3)]
        zipPartitionsDemo(sc)   // 对rdd按照分区进行合并
        zipWithIndexDemo(sc)    // 将RDD中的元素与索引进行合并
        zipWithUniqueIdDemo(sc) // 将RDD的元素与唯一的ID 进行拉链(zip)操作。
        sortByDemo(sc)  // 用于RDD 元素排序，ascending参数: 是否升序
        partitionByDemo(sc) // 只有当RDD中的元素为k/v形式时，才可以使用k/v转换操作。
        reduceByKeyDemo(sc) // 该操作将RDD[K,V]的元素按照相同的K 对应V进行聚合操作，其存在多重载形式，还可以设置新RDD的分区数
        groupByKeyDemo(sc)  // 将RDD 中每个键值对按照K对V进行汇总，把具有相同K的V放置在一个集合中
        combineByKeyDemo(sc)
        aggregateByKeyDemo(sc) // 实际使用场景 未知
        aggregateByKeyDemo01(sc)
        foldByKeyDemo(sc)   // 按照K进行汇聚，共计v的数值，然后根据zeroValue 对v值进行操作
        sortByKeyDemo(sc)   // 根据RDD key 进行排序
        joinDemo(sc)        // 对RDD[K,V]与RDD[K,W]进行join操作。按照相同的K，将V与M联合在一起，组成新的RDD[(K,(V,W))].
        cogroupDemo(sc)     // 像多 RDD 的 join
        mapValuesDemo(sc)   // 在Rdd[k,v] 中，依次对V进行操作，返回新的RDD
        flatMapValuesDemo(sc)   // 在RDD[K,V]中，对V进行flatMap操作，即依次遍历V类型的元素并应用f函数，最终返回一个集合
        sc.stop()
    }
    def sparkContextInit(): SparkContext ={
        // 初始化 sparkconf 对象，设置基本任务参数
        val conf = new SparkConf()
                // 设置目标Master 通信地址
                //.setMaster("yarn")
                .setMaster("local[1]")
                //设置任务名称
                .setAppName("TransformationsDemo")
        //实例化 SparkContext ，Spark 的对外接口负责用户与Spark内部的交互®通信
        return new SparkContext(conf)
    }

    /**
      *  def map[U:ClassTag](f:T => U):RDD[U]
      *    : 遍历该RDD中每一数据项，从而产生新的RDD
      */

    def mapDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize( 1 to 10)
        // 对rdd1 每个元素 进行  * 10 操作
        val rdd2 = rdd1.map(_ * 10)
        rdd1.foreach(println)
        rdd2.foreach(println)
    }

    /**
      * def flatMap[U:ClassTag](f:T => TraversableOnce[U]:RDD[U])
      *     : 对RDD中每一个元素通过应用f函数一次转换为新的元素，并封装到RDD中。
      *    意思就是如果有A;B;C;D;B;C字符串，则（A,B）,(C,D),(D,B)相邻字符对出现一次，(B,C)出现两次。
      */
    def flatMapDemo (sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array("A;B;C;D;B;D;C;B;D;A;E;D;C;A;B"))
        val rdd2 = rdd1.map(_.split(";"))
        val rdd3 = rdd2.flatMap( x=> {
            for(i <- 0 until x.length-1) yield (x(i)+","+x(i+1),1)
        }).reduceByKey(_+_)
        rdd3.foreach(println)
    }

    /**
      * def filter(f:T => Boolean):RDD[T]
      *     : 当对 RDD 调用 filter 方法时，会对每个元素应用f函数，如果返回值为true，则该元素会被添加到新的RDD中
      *  求出1-100间的素数
      *  素数: 一个大于1的自然数，除了1和它自身外，不能被其他自然数整除的数叫做质数
      */
    def filterDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 100)
        val rdd2 = rdd1.filter( n=>{
            // 如果n 小于 2 ， 那么 flag 返回false
            var flag = if(n<2) false else true
            //breakable scala 中 中断循环的操作
            breakable{
                for( x<-2 until n){
                    // 如果 2-100 间 n 能整除的数的话 flag 返回false ；不是素数
                    if(n % x == 0){
                        flag = false
                        break
                    }
                }
            }
            flag
        })
        println(rdd2.collect.mkString(","))
    }

    /**
      *  def distinct (numPartitions:Int) (implicit ord:Ordering[T] = null):RDD[T]
      *  def distinct ():RDD[T]
      *  会对内部元素进行去重，生成新的RDD，还可以通过numPartitions 参数设置新的RDD分区个数
      *  举例: RDD中用户数据按照 姓名 去重
      */
    def distinctDemo (sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array("Alice","Nick","Alice","Kotlin","Catalina","Catalina"),3)
        val rdd2 = rdd1.distinct
        println(rdd2.collect.mkString(","))
    }

    /**
      * def mapPartitions
      *     : map 操作是依次操作每个元素
      *     : mapPartitions 操作是分区，按照分区对数据进行批处理
      */
    def mapPartitionsDemo (sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("201800001",83),
            ("201800002",87),
            ("201800003",100),
            ("201800004",95),
            ("201800005",87)),3)
        val rdd2 = rdd1.mapPartitions(iter => {
            var result = List[String]()

            while (iter.hasNext){
                result = iter.next() match {
                        //如果 grade 列 大于95 则符合条件，添加到列表中
                    case (id,grade) if grade >= 95  => id + "_" + grade :: result
                    case  _ => result
                }
            }
            result.iterator
        })
        println(rdd2.collect.mkString(","))
    }

    /**
      * def mapPartitionsWithIndex
      *     : 在 mapPartitionsWith 方法上 添加了 显示数据在哪个分区
      */
    def mapPartitionsWithIndexDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("201800001",83),
            ("201800002",87),
            ("201800003",100),
            ("201800004",95),
            ("201800005",96),
            ("201800006",95),
            ("201800007",100)),2)
        val rdd2 = rdd1.mapPartitionsWithIndex((index,iter) =>{
            var result = List[String]()
            while (iter.hasNext) {
                result = iter.next() match {
                    case (id,grade) if grade >=95 => id + "_" +grade +"["+index +"]" :: result
                    case _ => result
                }
            }
            result.iterator
        })
        println(rdd2.collect.mkString(":"))
    }

    /**
      * def union(other:RDD[T]):RDD[T]
      *     : 对两个RDD进行并集运算，返回新的RDD
      */
    def unionDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 10 )
        val rdd2 = sc.parallelize(1 to 20 )
        val rdd3 = rdd1.union(rdd2)
        // 合并、去重、打印成一列
        println(rdd3.distinct.collect.mkString(":"))
    }

    /**
      * def intersection(other:RDD[T]):RDD[T]
      *     : 对两个RDD进行交集运算，返回新的RDD
      */
    def intersectionDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 10 )
        val rdd2 = sc.parallelize(1 to 20 )
        val rdd3 = rdd1.intersection(rdd2)
        println(rdd3.collect.mkString(":"))
    }

    /**
      * def subtract(other:RDD[T]):RDD[T]
      *     : 对两个RDD进行差集运算，返回新的RDD
      * 不会对结果进行去重
      */
    def subtractDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 10 )
        val rdd2 = sc.parallelize(1 to 20 )
        // 返回 rdd2 中存在 rdd1中不存在的元素、存储到rdd3中
        val rdd3 = rdd2.subtract(rdd1)
        println(rdd3.collect.mkString(":"))
    }

    /**
      *  def coalesce
      *     : 修改RDD 分区数; 根据实际分区的数据量 对分区数进行增、减 , 从而提高每个task处理的效率
      */
    def coalesceDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize( 1 to 100 , 10)
        println("rdd1 . partition = " + rdd1.partitions.length)
        val rdd2 = rdd1.coalesce(5)
        println("rdd2 . partition = " + rdd2.partitions.length)
        val rdd3 = rdd2.coalesce(7)
        println("rdd3 . partition = " + rdd3.partitions.length)
    }

    /**
      * def repartition
      *     : 内部执行的是coalesce,参数 shuffle 的默认值为true
      */
    def repartitionDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize( 1 to 100 , 10)
        println("rdd1 . partition = " + rdd1.partitions.length)
        val rdd2 = rdd1.repartition(5)
        println("rdd2 . partition = " + rdd2.partitions.length)
        val rdd3 = rdd2.repartition(7)
        println("rdd3 . partition = " + rdd3.partitions.length)
    }

    /**
      *  def randomSplit(
      *     weights:Array[Double],
      *     seed : Long = Utiles.random.nextLong) : Array[RDD[T]]
      *     该操作根据第一个参数weight对一个RDD进行拆分。拆分后产生几个RDD，取决于设置了几个权重值(weight)
      *     数据量分配与权重比例一致
      *  权重比例之和最好为1，Array(1,4,5),权重比例之和为 1/10+4/10+5/10 = 1
      */
    def randomSplitDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 10 ,3)
        val splitRdd = rdd1.randomSplit(Array(1,4,5))
        println(splitRdd(0).collect.mkString(",")+"\n"+splitRdd(1).collect.mkString(",")+"\n"+splitRdd(2).collect.mkString(","))

    }

    /**
      *  def glom():RDD[Array[T]]
      *  该操作将RDD 中每个分区变成一个数组，并放置在新的RDD中，数组中元素的类型与原分区中元素类型一致
      */
    def glomDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 10 , 5)
        val rdd2 = rdd1.glom()
        println(rdd2.collect.mkString(":"))
    }

    /**
      *  def zip[U:ClassTag](other:RDD[T]):RDD[(T,U)]
      *     该操作对两个RDD中的元素，以键值对的形式进行合并。其中键值对中的key为第一个RDD中的元素，键值对中的Value为第二个RDD中的元素。
      *  注: 在使用zip操作时，需要确保两个RDD中的元素个数与分区个数都完全一样，否则会出现异常
      */
    def zipDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 3 , 2)
        val rdd2 = sc.parallelize(Array("A","B","C"),2)
        val rdd3 = rdd1.zip(rdd2)
        println(rdd3.collect.mkString(","))
    }


    /**
      * def zipWithIndex() :RDD[(T,Long)]
      *     该操作将RDD中的元素与索引进行合并
      */
    def zipWithIndexDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array("A","B","C","D","E"),2)
        val rdd2 = rdd1.zipWithIndex()
        println(rdd2.collect.mkString(","))
    }

    /**
      *  def zipWithUniqueId():RDD[T,Long]
      *     该操作将RDD的元素与唯一的ID 进行拉链(zip)操作。
      *     与zipWithIndex 不通的是: 它不需要先通过计算生成"ZippedWithIndexRDD"
      */
    def zipWithUniqueIdDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array("A","B","C","D","E"),2)
        val rdd2 = rdd1.zipWithUniqueId()
        println(rdd2.collect.foreach(println))
    }

    /**
      *  def zipPartitions()
      *     第一个参数传入另一个RDD
      *     第二个参数"f函数"，用于的定义如何对每一个分区中的元素进行zip操作
      * rdd1 / rdd2 应用zipParttions操作，将两个RDD中的数据按照分区进行合并。
      */
    def zipPartitionsDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 10 , 2)
        val rdd2 = sc.parallelize(20 to 25 , 2)
        val rdd3 = rdd1.zipPartitions(rdd2)((rddIter1,rddIter2) => {
            var result = List[(Int, Int)]()
            while (rddIter1.hasNext && rddIter2.hasNext){
                // 这里可以自行定义方法
                result ::= (rddIter1.next(),rddIter2.next())
            }
            result.iterator
        })
        println(rdd3.collect.mkString(","))
    }

    /**
      * def sortBy() ascending参数: 是否升序
      *     该操作用于排序数据
      */
    def sortByDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("dog",3),
            ("cat",1),
            ("hadoop",2),
            ("spark",3),
            ("apple",2)
        ))
        val rdd2 = rdd1.sortBy(_._2,false)
        println(rdd2.collect.mkString(","))
    }

    /**
      * def partitionBy(partitioner:Partitioner):RDD[(K,V)]
      *     只有当RDD中的元素为k/v形式时，才可以使用k/v转换操作。
      */
    def partitionByDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("Alice",18),
            ("Bob",3),
            ("Thomas",20)
        ),2)
        // new HashPartitioner(6) 官网自带的哈希分区器
        val rdd2 = rdd1.partitionBy(new HashPartitioner(6))
        println(rdd2.collect.mkString(","))
    }

    /**
      * def reduceByKey(func:(V,V) => V):RDD[(K,V)]
      *     该操作将RDD[K,V]的元素按照相同的K 对应V进行聚合操作，其存在多重载形式，还可以设置新RDD的分区数
      *
      * 将班级中学生的考试分数按照"60分一下","60~79分","80~100分" 进行聚合，统计每个分段有多少学生
      */
    def reduceByKeyDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("Alice",95),
            ("Bob",37),
            ("Thomas",100),
            ("Catalina",77),
            ("Karen",87)
        ),2)
        val rdd2 = rdd1.map({
            case (_,grade) if grade <= 60 => ("C",1)
            case (_,grade) if grade > 60 && grade < 80 => ("B",1)
            case (_,grade) if grade >= 80 => ("A",1)
        })
        val rdd3 = rdd2.reduceByKey(_ + _)
        println(rdd3.collect.mkString(","))
    }

    /**
      * def groupByKey(): RDD [(K,Iterable[T])]
      *     该操作将RDD 中每个键值对按照K对V进行汇总，把具有相同K的V放置在一个集合中
      *     该操作可以制定分区器或者分区数(默认使用 HashPartitioner)
      */
    def groupByKeyDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("班级1","Alice"),
            ("班级2","Tom"),
            ("班级1","Catalina"),
            ("班级3","Murdoch"),
            ("班级2","Fisker")
        ),2)
        val rdd2 = rdd1.groupByKey()
        println(rdd2.collect.mkString(":"))
    }

    /**
      * def combineByKey [C](
      *     createCombiner:V => C,
      *     mergeValue:(C,V) =>C ,
      *     mergeCombiners:(C,C) => C
      *     ) :RDD[(K,C)]
      *     createCombiner : 遍历RDD中每一个元素时，如果该元素对应的K 第一次被访问，则会调用createCombiner 操作来创建组合器从而将元素的值从V类型转换为C类型
      *     mergeValue: 将C类型值与V类型值聚合为C类型
      *     mergeCombiners: 一个RDD可以有多个分区，多个拥有相同的键(Key)的元素可以被分散到不通分区中，每一个分区有独立的Combiner，拥有相同键的元素会先在各自所属分区内进行聚合，然后在通过mergeCombiners 函数将多个分区各自的聚合结果按照键，做最后的聚合
      *  举例 : 求每个班级的平均分输
      */
    def combineByKeyDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("班级1",95f),
            ("班级2",80f),
            ("班级1",75f),
            ("班级3",97f),
            ("班级2",88f)
        ),2)
        val rdd2 = rdd1.combineByKey(
            // 将("班级1",95f) 转换成C类型
            grade => (grade,1),
            // 将相同类型(C类型)聚合  grade聚合，人数聚合
            (gc:(Float,Int),grade) => (gc._1 + grade,gc._2 + 1),
            // 将分散的分区结果按照C类型 进行最后的聚合，得出最后的班级，平均分输
            (gc1:(Float,Int),gc2:(Float,Int)) => (gc1._1 + gc2._1,gc1._2+gc2._2)
        )
        val rdd3 = rdd2.map(t=> (t._1,t._2._1/t._2._2))
        println(rdd3.collect.mkString(":"))
    }

    /**
      * def aggregateByKey[U:Classtag](zeroValue:U)(seqOp:(U.V) => U,combOp:(U,U) => U):RDD[(K,V)]
      *     zeroValue:设置聚合时的初始值。不要认为初始值总是数字，集合也很常用
      *     seqOp: 将值V聚合到类型为U的对象中
      *     combOp: 跨分区聚合，对数据进行最终的汇总时，调用此操作
      *  举例: 求每个用户访问过的接口，且要求去重
      */
    def aggregateByKeyDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("用户1","接口1"),
            ("用户2","接口1"),
            ("用户1","接口1"),
            ("用户1","接口2"),
            ("用户2","接口3")
        ),2)
        val rdd2 = rdd1.aggregateByKey(collection.mutable.Set[String]())(
            (urlSet,url) => urlSet += url,
            (urlSet1,urlSet2) => urlSet1 ++= urlSet2
        )
        println(rdd2.collect.mkString(":"))

    }

    // RDD.aggregateByKey(0)(_+_,_+_) == RDD.reduceByKey(_+_)
    def aggregateByKeyDemo01(sc:SparkContext): Unit =
    {
        var z = sc.parallelize(List(
            ("cat",2),
            ("cat", 5),
            ("mouse", 4),
            ("cat", 12),
            ("dog", 12),
            ("mouse", 2)), 2)
        println(z.groupByKey().collect.mkString(":"))
        //打印出各partition的数据
        def myfunc(index: Int, iter: Iterator[(String, Int)]) : Iterator[(String, Int)] = {
            println("[Parition :" +  index + iter.toList + "]")
            iter
        }
        z.mapPartitionsWithIndex(myfunc).collect
        def myfunc1(a:Int, b:Int) : Int = {
            println("[myfunc1: " + a + "\t" + b + "]");
            a * b
        }
        def myfunc2(a:Int, b:Int) : Int = {
            println("[myfunc2: " + a + "\t" + b + "]");
            a + b
        }

        val rdd23 = z.aggregateByKey(10)(myfunc1, myfunc2).collect
        rdd23.foreach(println(_))
        val rdd2 = z.aggregateByKey(10)(_*_,_+_)
        rdd2.foreach(println(_))

        /**
          *
          *  执行原理分析 :
          * Parition 0: List((cat,2), (cat,5), (mouse,4))
          * Parition 1: List((cat,12), (dog,12), (mouse,2))
          * aggregate的初始值参数为：10。
          * 第一个reduce函数为：乘积。
          * 第二个reduce函数为：求和。
          *
          * Parition内：
          * Parition 0:
          * cat: 10(初始化) * 2 = 20
          * cat: 20(上一步计算结果) * 5 = 100
          * mouse: 10(初始值) * 4 = 40
          * Parition 1:
          * cat: 10(初始值) * 12 = 120
          * dog: 10(初始值) * 12 = 120
          * mouse: 10(初始值) * 2 = 20
          * 第二个reduce：
          * cat: 100(Partiion0结果) + 120(Partiion1结果) = 220
          * dog: 120(Partiion1结果) = 120
          * mouse: 40(Partiion0结果) + 20(Partiion1结果) = 60
          */
    }

    /**
      *  def foldByKey(zeroValue:V)(func:(V,V) => V) :RDD [(k,v)]    比 reduceByKey 多一个 初始值的功能
      *     按照K进行汇聚，共计v的数值，然后根据zeroValue 对v值进行操作
      *    举例: 会员打折应付金额=总金额-活动金额
      */
    def foldByKeyDemo(sc:SparkContext): Unit =
    {
         val rdd1 = sc.parallelize(Array(
             ("会员A",300f),
             ("会员A",100f),
             ("会员B",150f),
             ("会员B",500f),
             ("会员C",350f)
         ),2)
        val rdd2 = rdd1.foldByKey(-100f)(_+_)
        println(rdd2.collect.mkString(":"))
    }

    /**
      *  def sortByKey(ascending:Boolean=true,numPartitions:Int = self.partitions.length):RDD [(K.V)]
      *     RDD[K,V] 执行该操作，必须确保K实现了Ordering 接口，返回一个新的按照K排序RDD[K,V]集合。可以选择正序或倒序
      *  举例: 将词频统计的结果按照字母排列顺序输出
      */
    def sortByKeyDemo(sc:SparkContext): Unit =
    {
        /**
          *
            sortBy(_._2,false)
            (dog,3):(spark,3):(hadoop,2):(apple,2):(cat,1)

            sortByKey()
            (apple,2):(cat,1):(dog,3):(hadoop,2):(spark,3)
          */
        val rdd1 = sc.parallelize(Array(
            ("dog",3),
            ("cat",1),
            ("hadoop",2),
            ("spark",3),
            ("apple",2)))
        val rdd2 = rdd1.sortByKey()
        println(rdd2.collect.mkString(":"))
    }


    /**
      * def join[W](other:RDD[(K,W)],partitioner:Partitioner):RDD[(K,(V,W))]
      *     对RDD[K,V]与RDD[K,W]进行join操作。按照相同的K，将V与M联合在一起，组成新的RDD[(K,(V,W))].
      *  举例: 对RDD [姓名,年龄]集合与[姓名,性别]集合进行join操作.
      */
    def joinDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("Alice",19),
            ("Bob",20),
            ("Thomas",30),
            ("Catalina",25),
            ("Kotlin",27)
        ),2)

        val rdd2 = sc.parallelize(Array(
            ("Alice","female"),
            ("Bob","male"),
            ("Thomas","male"),
            ("Catalina","famale"),
            ("Kotlin","female")
        ),2)
        val rdd3 = rdd1.join(rdd2)
        println(rdd3.collect.mkString(":"))
    }

    /**
      * def cogroup[W1,W2,W3](
      *     other1:RDD[(K,W1)],
      *     other2:RDD[(K,W2)],
      *     other3:RDD[(K,W3)],
      *     partitioner:Partitioner
      *     ) : RDD [(K,(Iterable[V]),(Iterable[W1]),(Iterable[W2]),(Iterable[W3]))]
      *
      *  对 RDD 根据K进行关联操作，得到新的RDD 可传入1-3个RDD ，如果有无法关联的 K 返回空
      *  举例: 对 rdd[姓名，年龄] rdd[姓名，性别] rdd[姓名，地址] 进行关联操作。
      */
    def cogroupDemo(sc:SparkContext): Unit =
    {
        // 姓名，年龄
        val rdd1 = sc.parallelize(Array(
            ("Alice",19),
            ("Bob",20),
            ("Thomas",30),
            ("Catalina",25),
            ("Kotlin",27),
            ("Karen",99)
        ),2)
        // 姓名，性别
        val rdd2 = sc.parallelize(Array(
            ("Alice","female"),
            ("Bob","male"),
            ("Thomas","male"),
            ("Catalina","famale"),
            ("Kotlin","female")
        ),2)
        // 姓名，住址
        val rdd3 = sc.parallelize(Array(
            ("Alice","address1"),
            ("Bob","address2"),
            ("Thomas","address3"),
            ("Catalina","address4"),
            ("Kotlin","address5"),
            ("Karen","address6")
        ),2)
        // RDD 汇总
        val rdd4 = rdd1.cogroup(rdd2,rdd3,2)
        println(rdd4.collect.mkString(":"))

    }

    /**
      * def mapValues[U](f:V => U) :RDD [(K,U)]
      *   在Rdd[k,v] 中，依次对V进行操作，返回新的RDD
      *   举例: 在RDD[姓名，月收入] 集合中，对收入添加量词描述.
      */
    def mapValuesDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("Alice",3500),
            ("Bob",2000),
            ("Thomas",10000),
            ("Catalina",2000),
            ("Kotlin",4000),
            ("Karen",9000)
        ),2)
        val rdd2 = rdd1.mapValues( _ + "元")
        println(rdd2.collect.mkString(":"))
    }

    /**
      *  def flatMapValues[U](f:V => TraversableOnce[U]) :RDD[(K,V)]
      *     在RDD[K,V]中，对V进行flatMap操作，即依次遍历V类型的元素并应用f函数，最终返回一个集合
      *  举例: 对RDD[文件名称,某一行单词] 进行词频统计，并在结果中体现每个文件分别出现的单词个数。
      */
    def flatMapValuesDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("文件A","cat\tdog\thadoop\tcat"),
            ("文件A","cat\tdog\thadoop\tspark"),
            ("文件B","cat\tspark\thadoop\tcat"),
            ("文件B","spark\tdog\tspark\tcat")
        ),2)
        val rdd2 = rdd1.flatMapValues(_.split("\t"))
        val rdd3 = rdd2.map((_,1))
        val rdd4 = rdd3.reduceByKey(_+_).sortBy(_._1)
        // ((文件A,cat),3):((文件A,dog),2):((文件A,hadoop),2):((文件A,spark),1):((文件B,cat),3):((文件B,dog),1):((文件B,hadoop),1):((文件B,spark),3)
        println(rdd4.collect.mkString(":"))
    }
}



